package com.calabar.phm.etl.driver.spark;

import java.util.*;

/**
 * @author zmc
 * @version V1.0
 * @date 2016-3-1 10:22:03
 * @desc 存储算子关系（DAG）
 */
public final class DAG {
    /**
     * 根节点（输入算子）
     */
    private Set<Operator> roots = new LinkedHashSet<>();
    /**
     * 叶子节点（输出算子）
     */
    private Set<Operator> leaves = new LinkedHashSet<>();
    /**
     * 正向图（父子关系）
     */
    private Map<Operator, List<Operator>> workGraph = new HashMap<>();
    /**
     * 逆向图（子父关系）
     */
    private Map<Operator, List<Operator>> invertedWorkGraph = new HashMap<>();


    /**
     * 从叶子节点进行深度遍历，返回遍历顺序的拓扑图
     *
     * @return
     */
    public List<Operator> getAllWork() {
        List<Operator> result = new LinkedList<>();
        Set<Operator> seen = new HashSet<>();
        for (Operator leaf : getLeaves()) {
            visit(leaf, seen, result);
        }
        return result;
    }

    private void visit(Operator children, Set<Operator> seen, List<Operator> result) {
        if (seen.contains(children)) {
            return;
        }
        seen.add(children);
        for (Operator parent : getParents(children)) {
            if (!seen.contains(parent)) {
                visit(parent, seen, result);
            }
        }
        result.add(children);
    }

    public List<Operator> getParents(Operator str) {
        if (invertedWorkGraph.containsKey(str) && invertedWorkGraph.get(str) != null) {
            return invertedWorkGraph.get(str);
        } else {
            return new LinkedList<>(null);
        }
    }

    public List<Operator> getChildren(Operator str) {

        if (workGraph.containsKey(str) && workGraph.get(str) != null) {
            return workGraph.get(str);
        }
        return new LinkedList<>(null);
    }

    public void add(Operator operatorObj) {
        if (operatorObj == null || workGraph.containsKey(operatorObj)) {
            return;
        }
        workGraph.put(operatorObj, new LinkedList<Operator>());
        invertedWorkGraph.put(operatorObj, new LinkedList<Operator>());
        roots.add(operatorObj);
        leaves.add(operatorObj);
    }

    public void connnct(Operator a, Operator b) {
        if (a == null || b == null) {
            return;
        }
        workGraph.get(a).add(b);
        invertedWorkGraph.get(b).add(a);
        roots.remove(b);
        leaves.remove(a);
    }


    public Set<Operator> getLeaves() {
        return leaves;
    }

    public boolean isLeaf(Operator Operator) {
        return leaves.contains(Operator);
    }


    public boolean isRoot(Operator Operator) {
        return roots.contains(Operator);
    }

    @Override
    public String toString() {
        StringBuilder sb = new StringBuilder();

        sb.append("dag worgraph:")
                .append("\n");

        for (Map.Entry<Operator, List<Operator>> entry : workGraph.entrySet()) {
            sb.append(entry.getKey().getOperatorUnique())
                    .append(" --> ")
                    .append("【");

            for (Operator operator : entry.getValue()) {
                sb.append(operator.getOperatorUnique())
                        .append(" ");
            }
            sb.append("】").append("\n");
        }
        sb.append("dag topology:").append("\n");
        List<Operator> operators = getAllWork();
        int i = 0;
        for (Operator operator : operators) {
            i++;
            sb.append(operator.getOperatorUnique())
                    .append("【")
                    .append(i)
                    .append("】")
                    .append(" ");
        }
        return sb.toString();
    }
}
